// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: authentication-free shared key and token derivation between clients
// feature: support agent between domain
// feature: agent for vbs
// feature: provide query interfaces
// feature: develop dispatcher
// feature: content filter topic
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: add exception code
// feature: change idl check from md5 to crc16
// ------------------------------------------------------------------

#include <discovery/endpoint/EDP.h>

#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/ParticipantProxyData.h>

#include <discovery/participant/PDP.h>

#include <history/ReaderListener.h>
#include <edds/rtps/reader/RTPSReader.h>
#include <edds/rtps/writer/RTPSWriter.h>
#include <history/WriterListener.h>

#include <ertps/attributes/TopicAttributes.h>

#include <ertps/utils/StringMatching.h>

#include <ertps/types/TypeObjectFactory.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <mutex>

#include <elog/Log.hpp>
#include <edds/dds/core/policy/ParameterList.hpp>

#include <foonathan/memory/container.hpp>
#include <foonathan/memory/memory_pool.hpp>
#include <rtps/builtin/data/ProxyHashTables.hpp>
#include <rtps/network/ExternalLocatorsProcessor.hpp>
#include <utils/collections/node_size_helpers.hpp>

using namespace evbs::ertps;
using namespace evbs::ertps::types;
using evbs::edds::dds::PublicationMatchedStatus;
using evbs::edds::dds::SubscriptionMatchedStatus;
using ParameterList = evbs::edds::dds::ParameterList;

namespace evbs {
namespace ertps {
namespace rtps {

using reader_map_helper = utilities::collections::map_size_helper<GUID_t, SubscriptionMatchedStatus>;
using writer_map_helper = utilities::collections::map_size_helper<GUID_t, PublicationMatchedStatus>;

// static bool is_partition_empty(
//         const edds::dds::Partition_t& partition)
// {
//     return partition.size() <= 1 && 0 == strlen(partition.name());
// }

EDP::EDP(PDP* p, RTPSParticipantImpl* part) : mp_PDP(p), mp_RTPSParticipant(part) {}

EDP::~EDP() {
    // TODO Auto-generated destructor stub
}

bool EDP::newLocalReaderProxyData(RTPSReader* reader, const TopicAttributes& att, const ReaderQos& rqos,
                                  const vbsutil::xmlparser::ContentFilterProperty* content_filter) {
    logDebug(RTPS_EDP, "Adding " << reader->getGuid().entityId << " in topic " << att.topicName);

    reader->setIdlCrc16(att.getTopicCrc16Idl());
    reader->setTopicName(att.getTopicName().to_string());
    reader->register_peroid_metrics(att.getTopicName().to_string());

    auto init_fun = [this, reader, &att, &rqos, content_filter](ReaderProxyData* rpd, bool updating,
                                                                const ParticipantProxyData& participant_data,
                                                                RTPSParticipantImpl* localparticipant) {
        if (updating) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                      "Adding already existent reader " << reader->getGuid().entityId << " in topic " << att.topicName);
            return false;
        }

        const auto& ratt = reader->getAttributes();

        rpd->setIdlCrc16(reader->getIdlCrc16());
        rpd->isAlive(true);
        rpd->guid(reader->getGuid());
        rpd->key() = rpd->guid();
        if (ratt.multicastLocatorList.empty() && ratt.unicastLocatorList.empty()) {
            rpd->set_locators(participant_data.default_locators);
        } else {
            rpd->set_multicast_locators(ratt.multicastLocatorList, localparticipant);
            rpd->set_announced_unicast_locators(ratt.unicastLocatorList);
            evbs::edds::rtps::ExternalLocatorsProcessor::add_external_locators(*rpd, ratt.external_unicast_locators);
        }
        rpd->RTPSParticipantKey() = mp_RTPSParticipant->getGuid();
        rpd->topicName(att.getTopicName());
        rpd->typeName(att.getTopicDataType());
        rpd->topicKind(att.getTopicKind());
        rpd->m_qos.setQos(rqos, true);
        rpd->userDefinedId(static_cast<uint16_t>(ratt.getUserDefinedID()));
        rpd->type_plain(ratt.type_is_plain);

        if (nullptr != content_filter) {
            // Check content of ContentFilterProperty.
            if (!(0 < content_filter->content_filtered_topic_name.size() &&
                  0 < content_filter->related_topic_name.size() && 0 < content_filter->filter_class_name.size() &&
                  0 < content_filter->filter_expression.size())) {
                elogError(
                    RTPS_EDP, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                    "Content_filter config error " << reader->getGuid().entityId << " in topic " << att.topicName);
                return false;
            }

            rpd->content_filter(*content_filter);
        }

#if HAVE_SECURITY
        if (mp_RTPSParticipant->is_secure()) {
            rpd->security_attributes_ = ratt.security_attributes().mask();
            rpd->plugin_security_attributes_ = ratt.security_attributes().plugin_endpoint_attributes;
        } else {
            rpd->security_attributes_ = 0UL;
            rpd->plugin_security_attributes_ = 0UL;
        }
#endif  // if HAVE_SECURITY
        return true;
    };

    // ADD IT TO THE LIST OF READERPROXYDATA
    GUID_t participant_guid;
    ReaderProxyData* reader_data = this->mp_PDP->addReaderProxyData(reader->getGuid(), participant_guid, init_fun);
    if (reader_data == nullptr) {
        return false;
    }

    // PAIRING
    (void)pairing_reader_proxy_with_any_local_writer(participant_guid, reader_data);
    (void)pairingReader(reader, participant_guid, *reader_data);
    // DO SOME PROCESSING DEPENDING ON THE IMPLEMENTATION (SIMPLE OR STATIC)
    (void)processLocalReaderProxyData(reader, reader_data);
    if (reader->getSendMulti()) {
        mp_PDP->setSendMulti(true);
    }
    return true;
}

bool EDP::newLocalWriterProxyData(RTPSWriter* writer, const TopicAttributes& att, const WriterQos& wqos) {
    logDebug(RTPS_EDP, "Adding " << writer->getGuid().entityId << " in topic " << att.topicName);

    writer->setIdlCrc16(att.getTopicCrc16Idl());
    writer->setTopicName(att.getTopicName().to_string());
    writer->register_peroid_metrics(att.getTopicName().to_string());

    auto init_fun = [this, writer, &att, &wqos](WriterProxyData* wpd, bool updating,
                                                const ParticipantProxyData& participant_data,
                                                RTPSParticipantImpl* localparticipant) {
        if (updating) {
            logWarning(RTPS_EDP, "Adding already existent writer " << writer->getGuid().entityId << " in topic "
                                                                   << att.topicName);
            return false;
        }

        const auto& watt = writer->getAttributes();

        wpd->setIdlCrc16(writer->getIdlCrc16());
        wpd->guid(writer->getGuid());
        wpd->key() = wpd->guid();
        if (watt.multicastLocatorList.empty() && watt.unicastLocatorList.empty()) {
            wpd->set_locators(participant_data.default_locators);
        } else {
            wpd->set_multicast_locators(watt.multicastLocatorList, localparticipant);
            wpd->set_announced_unicast_locators(watt.unicastLocatorList);
            evbs::edds::rtps::ExternalLocatorsProcessor::add_external_locators(*wpd, watt.external_unicast_locators);
        }
        wpd->RTPSParticipantKey() = mp_RTPSParticipant->getGuid();
        wpd->topicName(att.getTopicName());
        wpd->typeName(att.getTopicDataType());
        wpd->topicKind(att.getTopicKind());
        wpd->typeMaxSerialized(writer->getTypeMaxSerialized());
        wpd->m_qos.setQos(wqos, true);
        wpd->userDefinedId(static_cast<uint16_t>(watt.getUserDefinedID()));
        wpd->persistence_guid(watt.persistence_guid);
#if HAVE_SECURITY
        if (mp_RTPSParticipant->is_secure()) {
            wpd->security_attributes_ = watt.security_attributes().mask();
            wpd->plugin_security_attributes_ = watt.security_attributes().plugin_endpoint_attributes;
        } else {
            wpd->security_attributes_ = 0UL;
            wpd->plugin_security_attributes_ = 0UL;
        }
#endif  // if HAVE_SECURITY
        return true;
    };

    // ADD IT TO THE LIST OF READERPROXYDATA
    GUID_t participant_guid;
    WriterProxyData* writer_data = this->mp_PDP->addWriterProxyData(writer->getGuid(), participant_guid, init_fun);
    if (writer_data == nullptr) {
        return false;
    }

    // PAIRING
    (void)pairing_writer_proxy_with_any_local_reader(participant_guid, writer_data);
    (void)pairingWriter(writer, participant_guid, *writer_data);
    // DO SOME PROCESSING DEPENDING ON THE IMPLEMENTATION (SIMPLE OR STATIC)
    (void)processLocalWriterProxyData(writer, writer_data);

    if (writer->getSendMulti()) {
        mp_PDP->setSendMulti(true);
    }
    return true;
}

bool EDP::updatedLocalReader(RTPSReader* reader, const TopicAttributes& att, const ReaderQos& rqos,
                             const vbsutil::xmlparser::ContentFilterProperty* content_filter, bool notify) {
    auto init_fun = [this, reader, &rqos, &att, content_filter](ReaderProxyData* rdata, bool updating,
                                                                const ParticipantProxyData& /*participant_data*/,
                                                                RTPSParticipantImpl* localparticipant) {
        // Should only be called for existent data
        (void)updating;
        assert(updating);

        if (reader->getAttributes().multicastLocatorList.empty() &&
            reader->getAttributes().unicastLocatorList.empty()) {
            rdata->set_announced_unicast_locators(localparticipant->getAttributes().defaultUnicastLocatorList);
        } else {
            rdata->set_multicast_locators(reader->getAttributes().multicastLocatorList, localparticipant);
            rdata->set_announced_unicast_locators(reader->getAttributes().unicastLocatorList);
        }
        rdata->m_qos.setQos(rqos, false);
        if (nullptr != content_filter) {
            // Check content of ContentFilterProperty.
            if (!(0 < content_filter->content_filtered_topic_name.size() &&
                  0 < content_filter->related_topic_name.size() && 0 < content_filter->filter_class_name.size() &&
                  0 < content_filter->filter_expression.size())) {
                return false;
            }

            rdata->content_filter(*content_filter);
        }
        rdata->isAlive(true);

        return true;
    };

    GUID_t participant_guid;
    ReaderProxyData* reader_data = this->mp_PDP->addReaderProxyData(reader->getGuid(), participant_guid, init_fun);
    if (reader_data != nullptr) {
        (void)processLocalReaderProxyData(reader, reader_data);
        (void)pairing_reader_proxy_with_any_local_writer(participant_guid, reader_data);
        (void)pairingReader(reader, participant_guid, *reader_data, notify);
        return true;
    }
    return false;
}

bool EDP::updatedLocalWriter(RTPSWriter* writer, const TopicAttributes& att, const WriterQos& wqos,
                             const vbsutil::xmlparser::ContentFilterProperty* content_filter, bool notify) {
    auto init_fun = [this, writer, &wqos, &att, content_filter](WriterProxyData* wdata, bool updating,
                                                                const ParticipantProxyData& /*participant_data*/,
                                                                RTPSParticipantImpl* localparticipant) {
        // Should only be called for existent data
        (void)updating;
        assert(updating);

        if (writer->getAttributes().multicastLocatorList.empty() &&
            writer->getAttributes().unicastLocatorList.empty()) {
            wdata->set_announced_unicast_locators(localparticipant->getAttributes().defaultUnicastLocatorList);
        } else {
            wdata->set_multicast_locators(writer->getAttributes().multicastLocatorList, localparticipant);
            wdata->set_announced_unicast_locators(writer->getAttributes().unicastLocatorList);
        }
        wdata->m_qos.setQos(wqos, false);
        return true;
    };

    GUID_t participant_guid;
    WriterProxyData* writer_data = this->mp_PDP->addWriterProxyData(writer->getGuid(), participant_guid, init_fun);
    if (writer_data != nullptr) {
        (void)processLocalWriterProxyData(writer, writer_data);
        (void)pairing_writer_proxy_with_any_local_reader(participant_guid, writer_data);
        /* 只适用于如下场景：“更新前qos不匹配导致没有match，此次更新后qos匹配能够match”，
           */
        (void)pairingWriter(writer, participant_guid, *writer_data, notify);
        return true;
    }
    return false;
}

bool EDP::unpairWriterProxy(const GUID_t& participant_guid, const GUID_t& writer_guid, bool removed_by_lease) {
    (void)participant_guid;

    logDebug(RTPS_EDP, writer_guid);

    (void)mp_RTPSParticipant->forEachUserReader([&, removed_by_lease](RTPSReader& r) -> bool {
        if (r.matched_writer_remove(writer_guid, removed_by_lease)) {
#if HAVE_SECURITY
            const GUID_t& reader_guid = r.getGuid();
            mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid, writer_guid);
#endif  // if HAVE_SECURITY

            // MATCHED AND ADDED CORRECTLY:
            ReaderListener* listener = nullptr;
            if (nullptr != (listener = r.getListener())) {
                MatchingInfo info;
                info.status = REMOVED_MATCHING;
                info.remoteEndpointGuid = writer_guid;
                listener->onReaderMatched(info);
            }
        }

        // traverse all
        return true;
    });

    return true;
}

bool EDP::unpairReaderProxy(const GUID_t& participant_guid, const GUID_t& reader_guid) {
    (void)participant_guid;

    logDebug(RTPS_EDP, reader_guid);

    (void)mp_RTPSParticipant->forEachUserWriter([&](RTPSWriter& w) -> bool {
        if (w.matched_reader_remove(reader_guid)) {
#if HAVE_SECURITY
            const GUID_t& writer_guid = w.getGuid();
            mp_RTPSParticipant->security_manager().remove_reader(writer_guid, participant_guid, reader_guid);
#endif  // if HAVE_SECURITY \
        // MATCHED AND ADDED CORRECTLY:
            WriterListener* listener = nullptr;
            if (nullptr != (listener = w.getListener())) {
                MatchingInfo info;
                info.status = REMOVED_MATCHING;
                info.remoteEndpointGuid = reader_guid;
                listener->onWriterMatched(info);

                // const PublicationMatchedStatus& pub_info =
                //     update_publication_matched_status(reader_guid, writer_guid, -1);
                // listener->onWriterMatched(pub_info);
            }
        }

        // traverse all
        return true;
    });

    return true;
}

bool EDP::valid_matching(const WriterProxyData* wdata, const ReaderProxyData* rdata, MatchingFailureMask& reason,
                         edds::dds::PolicyMask& incompatible_qos) {
    (void)reason.reset();
    (void)incompatible_qos.reset();

    if (wdata->topicName() != rdata->topicName()) {
        (void)reason.set(MatchingFailureMask::different_topic);
        logDebug(RTPS_EDP, "Data is different_topic.");
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_PROCESS) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_PROCESS)) &&
        wdata->guid().is_on_same_process_as(wdata->get_pid(), rdata->guid(), rdata->get_pid())) {
        logDebug(RTPS_EDP, "Intra process communication is disabled for writer of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_HOST) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_HOST)) &&
        wdata->guid().is_on_same_host_as(wdata->get_host_id(), rdata->guid(), rdata->get_host_id())) {
        logDebug(RTPS_EDP, "Intra host communication is disabled for writer of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTER_HOST) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTER_HOST)) &&
        !wdata->guid().is_on_same_host_as(wdata->get_host_id(), rdata->guid(), rdata->get_host_id())) {
        logDebug(RTPS_EDP, "Inter host communication is disabled for writer of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if ((wdata->m_qos.m_agent_flag & 1) != (rdata->m_qos.m_agent_flag & 1)) {
        logDebug(RTPS_EDP, "Writer agent flag is different. Topic:"
                               << wdata->topicName() << " local flag:" << (uint32_t)wdata->m_qos.m_agent_flag
                               << " remote flag:" << (uint32_t)rdata->m_qos.m_agent_flag);
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (wdata->topicKind() != rdata->topicKind()) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS:Remote Reader " << rdata->guid() << " is publishing in topic " << rdata->topicName()
                                                    << "(keyed:" << rdata->topicKind()
                                                    << "), local writer publishes as keyed: " << wdata->topicKind());

        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (!rdata->isAlive())  // Matching
    {
        elogError(RTPS_EDP, RetCode_t::RETCODE_PRECONDITION_NOT_MET, "ReaderProxyData object is NOT alive");

        return false;
    }

    if ((wdata->m_qos.m_reliability.kind == BEST_EFFORT_RELIABILITY_QOS) &&
        (rdata->m_qos.m_reliability.kind == RELIABLE_RELIABILITY_QOS))
    // Means our writer is BE but the reader wants RE
    {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):Remote Reader " << rdata->guid()
                                              << " is Reliable and local writer is BE ");
        (void)incompatible_qos.set(edds::dds::RELIABILITY_QOS_POLICY_ID);
    }

    if (wdata->m_qos.m_durability.kind < rdata->m_qos.m_durability.kind) {
        // TODO (MCC) Change log message
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):RemoteReader " << rdata->guid()
                                              << " has TRANSIENT_LOCAL DURABILITY and we offer VOLATILE");
        (void)incompatible_qos.set(edds::dds::DURABILITY_QOS_POLICY_ID);
    }

    if (wdata->m_qos.m_ownership.kind != rdata->m_qos.m_ownership.kind) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):Remote reader " << rdata->guid()
                                              << " has different Ownership Kind");
        (void)incompatible_qos.set(edds::dds::OWNERSHIP_QOS_POLICY_ID);
    }
#if 0
    if (wdata->m_qos.m_deadline.period > rdata->m_qos.m_deadline.period) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "):Remote reader " << rdata->guid()
                                              << " has smaller DEADLINE period" << wdata->m_qos.m_deadline.period
                                              << " > " << rdata->m_qos.m_deadline.period);
        (void)incompatible_qos.set(edds::dds::DEADLINE_QOS_POLICY_ID);
    }

    if (!wdata->m_qos.m_disablePositiveACKs.enabled && rdata->m_qos.m_disablePositiveACKs.enabled)
    {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                "Incompatible Disable Positive Acks QoS: writer is enabled but reader is not");
        incompatible_qos.set(edds::dds::DISABLEPOSITIVEACKS_QOS_POLICY_ID);
    }

    if (wdata->m_qos.m_liveliness.lease_duration > rdata->m_qos.m_liveliness.lease_duration) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "(topic: " << rdata->topicName()
                             << ") Incompatible liveliness lease durations: offered lease duration "
                             << wdata->m_qos.m_liveliness.lease_duration << " must be <= requested lease duration "
                             << rdata->m_qos.m_liveliness.lease_duration);
        (void)incompatible_qos.set(edds::dds::LIVELINESS_QOS_POLICY_ID);
    }

    if (wdata->m_qos.m_liveliness.kind < rdata->m_qos.m_liveliness.kind) {
        elogError(
            RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
            "(topic: " << rdata->topicName() << ") Incompatible liveliness kinds: offered kind is < requested kind");
        (void)incompatible_qos.set(edds::dds::LIVELINESS_QOS_POLICY_ID);
    }
#endif
    if (rdata->m_qos.m_e2eProtecion.needCheckEnableFlag || wdata->m_qos.m_e2eProtecion.needCheckEnableFlag) {
        if (wdata->m_qos.m_e2eProtecion.enabled != rdata->m_qos.m_e2eProtecion.enabled) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                      "(topic: " << rdata->topicName()
                                 << ") INCOMPATIBLE QOS m_e2eProtecion check err, wdata e2eProtecion flag = "
                                 << wdata->m_qos.m_e2eProtecion.enabled
                                 << ", rdata e2eProtecion flag= " << rdata->m_qos.m_e2eProtecion.enabled);
            return false;
        }
    }

#if HAVE_SECURITY
    // TODO: Check EndpointSecurityInfo
#endif  // if HAVE_SECURITY

    // Partition mismatch does not trigger status change
    if (incompatible_qos.any()) {
        (void)reason.set(MatchingFailureMask::incompatible_qos);
        return false;
    }
    logDebug(RTPS_EDP, "Type check. Type name:" << rdata->typeName() << " remote r:" << wdata->getIdlCrc16()
                                                << " local w: " << rdata->getIdlCrc16());

    // 走类型发现时候，本端类型crc值为空。使用对端crc值作为自己的。
    if (wdata->getIdlCrc16().empty()) {
        const_cast<WriterProxyData*>(wdata)->setIdlCrc16(rdata->getIdlCrc16());
    }
    if ((wdata->getIdlCrc16() != rdata->getIdlCrc16()) && !wdata->getIdlCrc16().empty() &&
        !rdata->getIdlCrc16().empty()) {
        elogWarning(RTPS_EDP, "data type different,"
                                  << "remote r:" << rdata->typeName() << " " << rdata->getIdlCrc16()
                                  << ", local w :" << wdata->typeName() << " " << wdata->getIdlCrc16());
    }

#if 0
    //Partition check:
    bool matched = false;
    if (wdata->m_qos.m_partition.empty() && rdata->m_qos.m_partition.empty())
    {
        matched = true;
    }
    else if (wdata->m_qos.m_partition.empty() && rdata->m_qos.m_partition.size() > 0)
    {
        for (auto rnameit = rdata->m_qos.m_partition.begin();
                rnameit != rdata->m_qos.m_partition.end(); ++rnameit)
        {
            if (is_partition_empty(*rnameit))
            {
                matched = true;
                break;
            }
        }
    }
    else if (wdata->m_qos.m_partition.size() > 0 && rdata->m_qos.m_partition.empty())
    {
        for (auto wnameit = wdata->m_qos.m_partition.begin();
                wnameit !=  wdata->m_qos.m_partition.end(); ++wnameit)
        {
            if (is_partition_empty(*wnameit))
            {
                matched = true;
                break;
            }
        }
    }
    else
    {
        for (auto wnameit = wdata->m_qos.m_partition.begin();
                wnameit !=  wdata->m_qos.m_partition.end(); ++wnameit)
        {
            for (auto rnameit = rdata->m_qos.m_partition.begin();
                    rnameit != rdata->m_qos.m_partition.end(); ++rnameit)
            {
                if (StringMatching::matchString(wnameit->name(), rnameit->name()))
                {
                    matched = true;
                    break;
                }
            }
            if (matched)
            {
                break;
            }
        }
    }
    if (!matched) //Different partitions
    {
        logWarning(RTPS_EDP, "INCOMPATIBLE QOS (topic: " << rdata->topicName() << "): Different Partitions");
        reason.set(MatchingFailureMask::partitions);
    }
#endif
    return true;
}

bool EDP::valid_matching(const ReaderProxyData* rdata, const WriterProxyData* wdata, MatchingFailureMask& reason,
                         edds::dds::PolicyMask& incompatible_qos) {
    (void)reason.reset();
    (void)incompatible_qos.reset();

    if (rdata->topicName() != wdata->topicName()) {
        (void)reason.set(MatchingFailureMask::different_topic);
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_PROCESS) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_PROCESS)) &&
        wdata->guid().is_on_same_process_as(wdata->get_pid(), rdata->guid(), rdata->get_pid())) {
        logDebug(RTPS_EDP, "Intra process communication is disabled for reader of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_HOST) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTRA_HOST)) &&
        wdata->guid().is_on_same_host_as(wdata->get_host_id(), rdata->guid(), rdata->get_host_id())) {
        logDebug(RTPS_EDP, "Intra host communication is disabled for reader of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (((wdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTER_HOST) ||
         (rdata->m_qos.m_ipc_mode & IpcMode::DISABLE_INTER_HOST)) &&
        !wdata->guid().is_on_same_host_as(wdata->get_host_id(), rdata->guid(), rdata->get_host_id())) {
        logDebug(RTPS_EDP, "Inter host communication is disabled for reader of " << wdata->topicName());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if ((wdata->m_qos.m_agent_flag & 1) != (rdata->m_qos.m_agent_flag & 1)) {
        logDebug(RTPS_EDP, "Reader gent flag is different. Topic:"
                               << wdata->topicName() << " local flag:" << (uint32_t)rdata->m_qos.m_agent_flag
                               << " remote flag:" << (uint32_t)wdata->m_qos.m_agent_flag);
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if (rdata->topicKind() != wdata->topicKind()) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS:Remote Writer " << wdata->guid() << " is publishing in topic " << wdata->topicName()
                                                    << "(keyed:" << wdata->topicKind()
                                                    << "), local reader subscribes as keyed: " << rdata->topicKind());
        (void)reason.set(MatchingFailureMask::inconsistent_topic);
        return false;
    }

    if ((rdata->m_qos.m_reliability.kind == RELIABLE_RELIABILITY_QOS) &&
        (wdata->m_qos.m_reliability.kind == BEST_EFFORT_RELIABILITY_QOS))
    // Means our reader is reliable but hte writer is not
    {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << wdata->topicName() << "): Remote Writer " << wdata->guid()
                                              << " is Best Effort and local reader is RELIABLE ");
        (void)incompatible_qos.set(edds::dds::RELIABILITY_QOS_POLICY_ID);
    }
    if (rdata->m_qos.m_durability.kind > wdata->m_qos.m_durability.kind) {
        // TODO (MCC) Change log message
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << wdata->topicName() << "):RemoteWriter " << wdata->guid()
                                              << " has VOLATILE DURABILITY and we want TRANSIENT_LOCAL";);
        (void)incompatible_qos.set(edds::dds::DURABILITY_QOS_POLICY_ID);
    }
    if (rdata->m_qos.m_ownership.kind != wdata->m_qos.m_ownership.kind) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << wdata->topicName() << "):Remote Writer " << wdata->guid()
                                              << " has different Ownership Kind");
        (void)incompatible_qos.set(edds::dds::OWNERSHIP_QOS_POLICY_ID);
    }
#if 0
    if (rdata->m_qos.m_deadline.period < wdata->m_qos.m_deadline.period) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "INCOMPATIBLE QOS (topic: " << wdata->topicName() << "):RemoteWriter " << wdata->guid()
                                              << "has smaller DEADLINE period:" << rdata->m_qos.m_deadline.period
                                              << " < " << wdata->m_qos.m_deadline.period);
        (void)incompatible_qos.set(edds::dds::DEADLINE_QOS_POLICY_ID);
    }

    if (rdata->m_qos.m_disablePositiveACKs.enabled && !wdata->m_qos.m_disablePositiveACKs.enabled)
    {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                "Incompatible Disable Positive Acks QoS: writer is enabled but reader is not");
        incompatible_qos.set(edds::dds::DISABLEPOSITIVEACKS_QOS_POLICY_ID);
    }

    if (wdata->m_qos.m_liveliness.lease_duration > rdata->m_qos.m_liveliness.lease_duration) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "(topic: " << wdata->topicName()
                             << ") Incompatible liveliness lease durations: offered lease duration "
                             << wdata->m_qos.m_liveliness.lease_duration << " must be <= requested lease duration "
                             << rdata->m_qos.m_liveliness.lease_duration);
        (void)incompatible_qos.set(edds::dds::LIVELINESS_QOS_POLICY_ID);
    }
    if (wdata->m_qos.m_liveliness.kind < rdata->m_qos.m_liveliness.kind) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "(topic: " << wdata->topicName()
                             << ") Incompatible liveliness kinds: offered kind is < than requested kind");
        (void)incompatible_qos.set(edds::dds::LIVELINESS_QOS_POLICY_ID);
    }
#endif
    if (rdata->m_qos.m_e2eProtecion.needCheckEnableFlag || wdata->m_qos.m_e2eProtecion.needCheckEnableFlag) {
        if (wdata->m_qos.m_e2eProtecion.enabled != rdata->m_qos.m_e2eProtecion.enabled) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                      "(topic: " << wdata->topicName()
                                 << ") INCOMPATIBLE QOS m_e2eProtecion check err, wdata e2eProtecion flag = "
                                 << wdata->m_qos.m_e2eProtecion.enabled
                                 << ", rdata e2eProtecion flag= " << rdata->m_qos.m_e2eProtecion.enabled);
            return false;
        }
    }

#if HAVE_SECURITY
    // TODO: Check EndpointSecurityInfo
#endif  // if HAVE_SECURITY

    // Partition mismatch does not trigger status change
    if (incompatible_qos.any()) {
        (void)reason.set(MatchingFailureMask::incompatible_qos);
        return false;
    }

    logDebug(RTPS_EDP, "Type check: Type name:" << rdata->typeName() << " remote w:" << wdata->getIdlCrc16()
                                                << " local r: " << rdata->getIdlCrc16());

    // 走类型发现时候，本端类型crc值为空。使用对端crc值作为自己的。
    if (rdata->getIdlCrc16().empty()) {
        const_cast<ReaderProxyData*>(rdata)->setIdlCrc16(wdata->getIdlCrc16());
    }

    if ((wdata->getIdlCrc16() != rdata->getIdlCrc16()) && !wdata->getIdlCrc16().empty() &&
        !rdata->getIdlCrc16().empty()) {
        elogWarning(RTPS_EDP, "data type different,"
                                  << "remote w:" << wdata->typeName() << " " << wdata->getIdlCrc16()
                                  << ", local r :" << rdata->typeName() << " " << rdata->getIdlCrc16());
    }

#if 0
    //Partition check:
    bool matched = false;
    if (rdata->m_qos.m_partition.empty() && wdata->m_qos.m_partition.empty())
    {
        matched = true;
    }
    else if (rdata->m_qos.m_partition.empty() && wdata->m_qos.m_partition.size() > 0)
    {
        for (auto rnameit = wdata->m_qos.m_partition.begin();
                rnameit != wdata->m_qos.m_partition.end(); ++rnameit)
        {
            if (is_partition_empty(*rnameit))
            {
                matched = true;
                break;
            }
        }
    }
    else if (rdata->m_qos.m_partition.size() > 0 && wdata->m_qos.m_partition.empty())
    {
        for (auto wnameit = rdata->m_qos.m_partition.begin();
                wnameit !=  rdata->m_qos.m_partition.end(); ++wnameit)
        {
            if (is_partition_empty(*wnameit))
            {
                matched = true;
                break;
            }
        }
    }
    else
    {
        for (auto wnameit = rdata->m_qos.m_partition.begin();
                wnameit !=  rdata->m_qos.m_partition.end(); ++wnameit)
        {
            for (auto rnameit = wdata->m_qos.m_partition.begin();
                    rnameit != wdata->m_qos.m_partition.end(); ++rnameit)
            {
                if (StringMatching::matchString(wnameit->name(), rnameit->name()))
                {
                    matched = true;
                    break;
                }
            }
            if (matched)
            {
                break;
            }
        }
    }
    if (!matched) //Different partitions
    {
        logWarning(RTPS_EDP, "INCOMPATIBLE QOS (topic: " <<  wdata->topicName() <<
                "): Different Partitions");
        reason.set(MatchingFailureMask::partitions);
    }
#endif

    logDebug(RTPS_EDP,
             "Remote IDL file crc16 check OK wcrc = " << wdata->getIdlCrc16() << "rcrc = " << rdata->getIdlCrc16());
    return true;
}

ProxyPool<ReaderProxyData>& EDP::get_temporary_reader_proxies_pool() {
    assert(mp_PDP != nullptr);
    return mp_PDP->get_temporary_reader_proxies_pool();
}

ProxyPool<WriterProxyData>& EDP::get_temporary_writer_proxies_pool() {
    assert(mp_PDP != nullptr);
    return mp_PDP->get_temporary_writer_proxies_pool();
}

// TODO This four functions share common code (2 to 2) and surely can be templatized.

bool EDP::pairingReader(RTPSReader* R, const GUID_t& participant_guid, const ReaderProxyData& rdata, bool notify) {
    (void)participant_guid;

    logDebug(RTPS_EDP, rdata.guid() << " in topic: \"" << rdata.topicName() << "\"");
    std::lock_guard<std::recursive_mutex> pguard(*mp_PDP->getMutex());

    for (ResourceLimitedVector<ParticipantProxyData*>::const_iterator pit = mp_PDP->ParticipantProxiesBegin();
         pit != mp_PDP->ParticipantProxiesEnd(); ++pit) {
        for (auto& pair : *(*pit)->m_writers) {
            WriterProxyData* wdatait = pair.second;
            MatchingFailureMask no_match_reason;
            edds::dds::PolicyMask incompatible_qos;
            bool valid = valid_matching(&rdata, wdatait, no_match_reason, incompatible_qos);
            const GUID_t& reader_guid = R->getGuid();
            const GUID_t& writer_guid = wdatait->guid();

            if (valid) {
                /*此处实际更新的是reader属性，但是调用reader的matched_writer_add，内部判断认为是writer属性变化，
                  并向上通知。正确的做法：本地reader更新时进入此流程应该和该writer原来没有match，本地reader更新属性后，
                  和发现的writer可以匹配，进行第一次匹配。如果和该writer已经匹配过，不应该做任何处理*/
                if (!R->matched_writer_is_matched(wdatait->guid())) {
#if HAVE_SECURITY
                    logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                    if (mp_RTPSParticipant->security_manager().is_client &&
                        mp_RTPSParticipant->security_manager().discovered_writer_without_handshake(
                            R->m_guid, (*pit)->m_guid, *wdatait, R->getAttributes().security_attributes(), false)) {
                        logDebug(RTPS_EDP, "Security manager returns true for reader " << reader_guid);
                    } else if (!mp_RTPSParticipant->security_manager().discovered_writer(
                                   R->m_guid, (*pit)->m_guid, *wdatait, R->getAttributes().security_attributes())) {
                        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                                  "Security manager returns an error for reader " << reader_guid);
                    }
#else
                    if (R->matched_writer_add(*wdatait)) {
                        logDebug(RTPS_EDP_MATCH, "WP:" << wdatait->guid() << " match R:" << R->getGuid()
                                                       << ". RLoc:" << wdatait->remote_locators());
                        // MATCHED AND ADDED CORRECTLY:
                        if (notify && R->getListener() != nullptr) {
                            MatchingInfo info;
                            info.status = MATCHED_MATCHING;
                            info.remoteEndpointGuid = writer_guid;
                            R->getListener()->onReaderMatched(info);
                            (void)reader_guid;
                        }
                    }
#endif  // if HAVE_SECURITY
                }
            } else {
                if (notify && no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
                    (R->getListener() != nullptr)) {
                    R->getListener()->on_requested_incompatible_qos(incompatible_qos);
                }

                if (R->matched_writer_is_matched(wdatait->guid()) && R->matched_writer_remove(wdatait->guid())) {
#if HAVE_SECURITY
                    mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid,
                                                                         wdatait->guid());
#else
                    (void)reader_guid;
#endif  // if HAVE_SECURITY

                    // MATCHED AND ADDED CORRECTLY:
                    if (notify && R->getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = REMOVED_MATCHING;
                        info.remoteEndpointGuid = writer_guid;
                        R->getListener()->onReaderMatched(info);
                    }
                }
            }
        }
    }
    return true;
}

// TODO Add a WriterProxyData argument with a nullptr default value
bool EDP::pairingWriter(RTPSWriter* W, const GUID_t& participant_guid, const WriterProxyData& wdata, bool notify) {
    (void)participant_guid;

    logDebug(RTPS_EDP, W->getGuid() << " in topic: \"" << wdata.topicName() << "\"");
    std::lock_guard<std::recursive_mutex> pguard(*mp_PDP->getMutex());

    for (ResourceLimitedVector<ParticipantProxyData*>::const_iterator pit = mp_PDP->ParticipantProxiesBegin();
         pit != mp_PDP->ParticipantProxiesEnd(); ++pit) {
        for (auto& pair : *(*pit)->m_readers) {
            ReaderProxyData* rdatait = pair.second;
            const GUID_t& reader_guid = rdatait->guid();
            if (reader_guid == c_Guid_Unknown) {
                continue;
            }

            MatchingFailureMask no_match_reason;
            edds::dds::PolicyMask incompatible_qos;
            bool valid = valid_matching(&wdata, rdatait, no_match_reason, incompatible_qos);
            if (valid) {
                /*此处实际更新的是writer属性，但是调用writer的matched_reader_add，内部判断认为是reader属性变化，
                  并向上通知。正确的做法：本地writer更新时进入此流程应该和该reader原来没有match，writer更新属性后，
                  和发现的reader可以匹配，进行第一次匹配。如果和该reader已经匹配过，不应该做任何处理*/
                if (!W->matched_reader_is_matched(reader_guid)) {
#if HAVE_SECURITY
                    logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                    if (mp_RTPSParticipant->security_manager().is_client &&
                        mp_RTPSParticipant->security_manager().discovered_reader_without_handshake(
                            W->getGuid(), (*pit)->m_guid, *rdatait, W->getAttributes().security_attributes(), false)) {
                        logDebug(RTPS_EDP, "Security manager returns true for writer " << W->getGuid());
                    } else if (!mp_RTPSParticipant->security_manager().discovered_reader(
                                   W->getGuid(), (*pit)->m_guid, *rdatait, W->getAttributes().security_attributes())) {
                        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                                  "Security manager returns an error for writer " << W->getGuid());
                    }
#else
                    if (W->matched_reader_add(*rdatait)) {
                        logDebug(RTPS_EDP_MATCH, "RP:" << rdatait->guid() << " match W:" << W->getGuid()
                                                       << ". WLoc:" << rdatait->remote_locators());
                        // MATCHED AND ADDED CORRECTLY:
                        if (notify && W->getListener() != nullptr) {
                            MatchingInfo info;
                            info.status = MATCHED_MATCHING;
                            info.remoteEndpointGuid = reader_guid;
                            W->getListener()->onWriterMatched(info);

                            // const GUID_t& writer_guid = W->getGuid();
                            // const PublicationMatchedStatus& pub_info =
                            //     update_publication_matched_status(reader_guid, writer_guid, 1);
                            // W->getListener()->onWriterMatched(pub_info);
                        }
                    }
#endif  // if HAVE_SECURITY
                }
            } else {
                if (notify && no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
                    (W->getListener() != nullptr)) {
                    W->getListener()->on_offered_incompatible_qos(incompatible_qos);
                }

                if (W->matched_reader_is_matched(reader_guid) && W->matched_reader_remove(reader_guid)) {
#if HAVE_SECURITY
                    mp_RTPSParticipant->security_manager().remove_reader(W->getGuid(), participant_guid, reader_guid);
#endif  // if HAVE_SECURITY \
        // MATCHED AND ADDED CORRECTLY:
                    if (notify && W->getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = REMOVED_MATCHING;
                        info.remoteEndpointGuid = reader_guid;
                        W->getListener()->onWriterMatched(info);  //LCOV_EXCL_STOP

                        // const GUID_t& writer_guid = W->getGuid();
                        // const PublicationMatchedStatus& pub_info =
                        //     update_publication_matched_status(reader_guid, writer_guid, -1);
                        // W->getListener()->onWriterMatched(pub_info);
                    }
                }
            }
        }
    }
    return true;
}

bool EDP::pairing_reader_proxy_with_any_local_writer(const GUID_t& participant_guid, ReaderProxyData* rdata) {
    (void)participant_guid;

    logDebug(RTPS_EDP, rdata->guid() << " in topic: \"" << rdata->topicName() << "\"");

    (void)mp_RTPSParticipant->forEachUserWriter([&, rdata](RTPSWriter& w) -> bool {
        auto temp_writer_proxy_data = get_temporary_writer_proxies_pool().get();
        GUID_t writerGUID = w.getGuid();

        if (mp_PDP->lookupWriterProxyData(writerGUID, *temp_writer_proxy_data)) {
            MatchingFailureMask no_match_reason;
            edds::dds::PolicyMask incompatible_qos;

            bool valid = valid_matching(temp_writer_proxy_data.get(), rdata, no_match_reason, incompatible_qos);
            const GUID_t& reader_guid = rdata->guid();

            temp_writer_proxy_data.reset();

            if (valid) {
#if HAVE_SECURITY
                logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                if (mp_RTPSParticipant->security_manager().is_client &&
                    mp_RTPSParticipant->security_manager().discovered_reader_without_handshake(
                        writerGUID, participant_guid, *rdata, w.getAttributes().security_attributes(), false)) {
                    logDebug(RTPS_EDP, "Security manager returns true for writer " << writerGUID);
                } else if (!mp_RTPSParticipant->security_manager().discovered_reader(
                               writerGUID, participant_guid, *rdata, w.getAttributes().security_attributes())) {
                    elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                              "Security manager returns an error for writer " << writerGUID);
                }
#else
                if (w.matched_reader_add(*rdata)) {
                    logDebug(RTPS_EDP_MATCH, "RP:" << rdata->guid() << " match W:" << w.getGuid()
                                                   << ". RLoc:" << rdata->remote_locators());
                    // MATCHED AND ADDED CORRECTLY:
                    if (w.getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = MATCHED_MATCHING;
                        info.remoteEndpointGuid = reader_guid;
                        w.getListener()->onWriterMatched(info);

                        // const PublicationMatchedStatus& pub_info =
                        //     update_publication_matched_status(reader_guid, writerGUID, 1);
                        // w.getListener()->onWriterMatched(pub_info);
                    }
                }
#endif  // if HAVE_SECURITY
            } else {
                if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (w.getListener() != nullptr)) {
                    w.getListener()->on_offered_incompatible_qos(incompatible_qos);
                }

                if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid)) {
#if HAVE_SECURITY
                    mp_RTPSParticipant->security_manager().remove_reader(w.getGuid(), participant_guid, reader_guid);
#endif  // if HAVE_SECURITY \
        // MATCHED AND ADDED CORRECTLY:
                    if (w.getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = REMOVED_MATCHING;
                        info.remoteEndpointGuid = reader_guid;
                        w.getListener()->onWriterMatched(info);

                        // const PublicationMatchedStatus& pub_info =
                        //     update_publication_matched_status(reader_guid, writerGUID, -1);

                        // w.getListener()->onWriterMatched(pub_info);
                    }
                }
            }
        }
        // next iteration
        return true;
    });

    return true;
}

#if HAVE_SECURITY
bool EDP::pairing_reader_proxy_with_local_writer(const GUID_t& local_writer,
                                                 const GUID_t& remote_participant_guid,  //LCOV_EXCL_START
                                                 ReaderProxyData& rdata) {
    logDebug(RTPS_EDP, rdata.guid() << " in topic: \"" << rdata.topicName() << "\"");

    (void)mp_RTPSParticipant->forEachUserWriter([&](RTPSWriter& w) -> bool {
        GUID_t writerGUID = w.getGuid();
        const GUID_t& reader_guid = rdata.guid();

        if (local_writer == writerGUID) {
            auto temp_writer_proxy_data = get_temporary_writer_proxies_pool().get();

            if (mp_PDP->lookupWriterProxyData(writerGUID, *temp_writer_proxy_data)) {
                MatchingFailureMask no_match_reason;
                edds::dds::PolicyMask incompatible_qos;
                bool valid = valid_matching(temp_writer_proxy_data.get(), &rdata, no_match_reason, incompatible_qos);

                temp_writer_proxy_data.reset();

                if (valid) {
                    logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                    if (mp_RTPSParticipant->security_manager().is_client &&
                        mp_RTPSParticipant->security_manager().discovered_reader_without_handshake(
                            writerGUID, remote_participant_guid, rdata, w.getAttributes().security_attributes(),
                            false)) {
                        logDebug(RTPS_EDP, "Security manager returns true for writer " << writerGUID);
                    } else if (!mp_RTPSParticipant->security_manager().discovered_reader(
                                   writerGUID, remote_participant_guid, rdata,
                                   w.getAttributes().security_attributes())) {
                        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                                  "Security manager returns an error for writer " << writerGUID);
                    }
                } else {
                    if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (w.getListener() != nullptr)) {
                        w.getListener()->on_offered_incompatible_qos(incompatible_qos);
                    }

                    if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid)) {
                        mp_RTPSParticipant->security_manager().remove_reader(w.getGuid(), remote_participant_guid,
                                                                             reader_guid);
                        // MATCHED AND ADDED CORRECTLY:
                        if (w.getListener() != nullptr) {
                            MatchingInfo info;
                            info.status = REMOVED_MATCHING;
                            info.remoteEndpointGuid = reader_guid;
                            w.getListener()->onWriterMatched(info);

                            // const PublicationMatchedStatus& pub_info =
                            //     update_publication_matched_status(reader_guid, writerGUID, -1);
                            // w.getListener()->onWriterMatched(pub_info);
                        }
                    }
                }
            }
        }
        // next iteration
        return true;
    });

    return true;
}  //LCOV_EXCL_STOP

bool EDP::pairing_remote_reader_with_local_writer_after_security(const GUID_t& local_writer,
                                                                 const ReaderProxyData& remote_reader_data) {
    bool matched = false;
    bool found = false;

    (void)mp_RTPSParticipant->forEachUserWriter([&](RTPSWriter& w) -> bool {
        GUID_t writerGUID = w.getGuid();

        const GUID_t& reader_guid = remote_reader_data.guid();

        if (local_writer == writerGUID) {
            found = true;

            if (w.matched_reader_add(const_cast<ReaderProxyData&>(remote_reader_data))) {
                logDebug(RTPS_EDP, "Valid Matching to local writer: " << writerGUID.entityId);

                matched = true;

                // MATCHED AND ADDED CORRECTLY:
                if (w.getListener() != nullptr) {
                    MatchingInfo info;
                    info.status = MATCHED_MATCHING;
                    info.remoteEndpointGuid = reader_guid;
                    w.getListener()->onWriterMatched(info);

                    // const PublicationMatchedStatus& pub_info =
                    //     update_publication_matched_status(reader_guid, writerGUID, 1);
                    // w.getListener()->onWriterMatched(pub_info);
                }
            }
            // don't look anymore
            return false;
        }
        // keep looking
        return true;
    });

    return found ? matched
                 : pairing_remote_reader_with_local_builtin_writer_after_security(local_writer, remote_reader_data);
}

#endif  // if HAVE_SECURITY

bool EDP::pairing_writer_proxy_with_any_local_reader(const GUID_t& participant_guid, WriterProxyData* wdata) {
    (void)participant_guid;

    logDebug(RTPS_EDP, wdata->guid() << " in topic: " << wdata->topicName());

    (void)mp_RTPSParticipant->forEachUserReader([&, wdata](RTPSReader& r) -> bool {
        auto temp_reader_proxy_data = get_temporary_reader_proxies_pool().get();
        GUID_t readerGUID = r.getGuid();

        if (mp_PDP->lookupReaderProxyData(readerGUID, *temp_reader_proxy_data)) {
            MatchingFailureMask no_match_reason;
            edds::dds::PolicyMask incompatible_qos;
            bool valid = valid_matching(temp_reader_proxy_data.get(), wdata, no_match_reason, incompatible_qos);
            const GUID_t& writer_guid = wdata->guid();

            temp_reader_proxy_data.reset();

            if (valid) {
#if HAVE_SECURITY
                logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                if (mp_RTPSParticipant->security_manager().is_client &&
                    mp_RTPSParticipant->security_manager().discovered_writer_without_handshake(
                        readerGUID, participant_guid, *wdata, r.getAttributes().security_attributes(), false)) {
                    logDebug(RTPS_EDP, "Security manager returns true for reader " << readerGUID);
                } else if (!mp_RTPSParticipant->security_manager().discovered_writer(
                               readerGUID, participant_guid, *wdata, r.getAttributes().security_attributes())) {
                    elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                              "Security manager returns an error for reader " << readerGUID);
                }
#else
                if (r.matched_writer_add(*wdata)) {
                    logDebug(RTPS_EDP_MATCH, "WP:" << wdata->guid() << " match R:" << r.getGuid()
                                                   << ". WLoc:" << wdata->remote_locators());
                    // MATCHED AND ADDED CORRECTLY:
                    if (r.getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = MATCHED_MATCHING;
                        info.remoteEndpointGuid = writer_guid;
                        r.getListener()->onReaderMatched(info);
                    }
                }
#endif  // if HAVE_SECURITY
            } else {
                if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (r.getListener() != nullptr)) {
                    r.getListener()->on_requested_incompatible_qos(incompatible_qos);
                }

                if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid)) {
#if HAVE_SECURITY
                    mp_RTPSParticipant->security_manager().remove_writer(readerGUID, participant_guid, writer_guid);
#endif  // if HAVE_SECURITY \
        // MATCHED AND ADDED CORRECTLY:
                    if (r.getListener() != nullptr) {
                        MatchingInfo info;
                        info.status = REMOVED_MATCHING;
                        info.remoteEndpointGuid = writer_guid;
                        r.getListener()->onReaderMatched(info);
                    }
                }
            }
        }
        // keep looking
        return true;
    });

    return true;
}

#if HAVE_SECURITY
bool EDP::pairing_writer_proxy_with_local_reader(const GUID_t& local_reader,
                                                 const GUID_t& remote_participant_guid,  //LCOV_EXCL_START
                                                 WriterProxyData& wdata) {
    logDebug(RTPS_EDP, wdata.guid() << " in topic: \"" << wdata.topicName() << "\"");

    (void)mp_RTPSParticipant->forEachUserReader([&](RTPSReader& r) -> bool {
        GUID_t readerGUID = r.getGuid();

        if (local_reader == readerGUID) {
            auto temp_reader_proxy_data = get_temporary_reader_proxies_pool().get();

            if (mp_PDP->lookupReaderProxyData(readerGUID, *temp_reader_proxy_data)) {
                MatchingFailureMask no_match_reason;
                edds::dds::PolicyMask incompatible_qos;
                bool valid = valid_matching(temp_reader_proxy_data.get(), &wdata, no_match_reason, incompatible_qos);
                const GUID_t& writer_guid = wdata.guid();

                temp_reader_proxy_data.reset();

                if (valid) {
                    logDebug(RTPS_EDP, "participant is client:" << mp_RTPSParticipant->security_manager().is_client);
                    if (mp_RTPSParticipant->security_manager().is_client &&
                        mp_RTPSParticipant->security_manager().discovered_writer_without_handshake(
                            readerGUID, remote_participant_guid, wdata, r.getAttributes().security_attributes(),
                            false)) {
                        logDebug(RTPS_EDP, "Security manager returns true for reader " << readerGUID);
                    } else if (!mp_RTPSParticipant->security_manager().discovered_writer(
                                   readerGUID, remote_participant_guid, wdata,
                                   r.getAttributes().security_attributes())) {
                        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                                  "Security manager returns an error for reader " << readerGUID);
                    }
                } else {
                    if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (r.getListener() != nullptr)) {
                        r.getListener()->on_requested_incompatible_qos(incompatible_qos);
                    }

                    if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid)) {
                        mp_RTPSParticipant->security_manager().remove_writer(readerGUID, remote_participant_guid,
                                                                             writer_guid);
                        // MATCHED AND ADDED CORRECTLY:
                        if (r.getListener() != nullptr) {
                            MatchingInfo info;
                            info.status = REMOVED_MATCHING;
                            info.remoteEndpointGuid = writer_guid;
                            r.getListener()->onReaderMatched(info);
                        }
                    }
                }
            }
            // don't keep searching
            return false;
        }
        // keep searching
        return true;
    });

    return true;
}  //LCOV_EXCL_STOP

bool EDP::pairing_remote_writer_with_local_reader_after_security(const GUID_t& local_reader,
                                                                 const WriterProxyData& remote_writer_data) {
    bool matched = false;
    bool found = false;

    (void)mp_RTPSParticipant->forEachUserReader([&](RTPSReader& r) -> bool {
        GUID_t readerGUID = r.getGuid();

        const GUID_t& writer_guid = remote_writer_data.guid();

        if (local_reader == readerGUID) {
            found = true;

            // TODO(richiware) Implement and use move with attributes
            if (r.matched_writer_add(const_cast<WriterProxyData&>(remote_writer_data))) {
                logDebug(RTPS_EDP, "Valid Matching to local reader: " << readerGUID.entityId);

                matched = true;

                // MATCHED AND ADDED CORRECTLY:
                if (r.getListener() != nullptr) {
                    MatchingInfo info;
                    info.status = MATCHED_MATCHING;
                    info.remoteEndpointGuid = writer_guid;
                    r.getListener()->onReaderMatched(info);
                }
            }
            // dont' look anymore
            return false;
        }
        // keep looking
        return true;
    });

    return found ? matched
                 : pairing_remote_writer_with_local_builtin_reader_after_security(local_reader, remote_writer_data);
}

#endif  // if HAVE_SECURITY

}  // namespace rtps
}  // namespace ertps
}  // namespace evbs
